package com.faxsun.web.aspect;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.annotation.PostConstruct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.faxsun.core.persist.entity.TracePO;
import com.faxsun.core.service.TraceService;

/**
 * Async persist trace data to db on a configurable interval 
 * with configurable batch size
 * @author leima
 *
 */
@Component
public class TraceAsyncHandler{    

    /** check interval in milliseconds, default 30 seconds*/
    public static final long CHK_INTERVAL = 30000;
    
    /** persist batch size, default 500*/
    public static final int BATCH_SIZE = 500;
    
    /** worker pool size, default 1*/
    public static final int POOL_SIZE = 1;
    
    private static AtomicBoolean start = new AtomicBoolean(false);

    @Value("${trace.async.interval}")
    private String checkIntervalStr = null;
    
    @Value("${trace.async.batchsize}")
    private String batchSizeStr = null;
    
    @Value("${trace.async.worker}")
    private String poolSizeStr = null;
    
    private long checkInterval = CHK_INTERVAL;    
    private int poolsize = POOL_SIZE;
    private int batchSize = BATCH_SIZE;
    
    public static ConcurrentLinkedQueue<TracePO> traceQueue = new ConcurrentLinkedQueue<TracePO>();     
    
    public static ScheduledExecutorService scheduler = null;
    
    @Autowired
    private TraceService traceService;
        
    @PostConstruct
    public void start() {
        
        if(checkIntervalStr != null){
           checkInterval = Long.parseLong(checkIntervalStr);
        }
        
        if(batchSizeStr != null){
           batchSize = Integer.parseInt(batchSizeStr);
        }
        
        if(poolSizeStr != null){
           poolsize = Integer.parseInt(poolSizeStr);
        }
        
        scheduler = Executors.newScheduledThreadPool(poolsize);
        
        if(!start.get()){            
            final Runnable persister = new Runnable() {
                public void run() { 
                    List<TracePO> traces = new ArrayList<TracePO>(batchSize);
                    
                    while(!traceQueue.isEmpty()){
                      TracePO trace = traceQueue.poll();
                      if(trace != null){
                         traces.add(trace);
                      }
                      
                      if(traces.size() == batchSize){
                         traceService.batchSave(traces); 
                         traces.clear();
                      }
                    }
                    
                    if(traces.size() > 0){
                      traceService.batchSave(traces);
                    }
                }
              };
              
              scheduler.scheduleAtFixedRate(persister, checkInterval, checkInterval, TimeUnit.MILLISECONDS);   
              
              start.compareAndSet(false, true);
        }
    }
    
}